package com.sduept.component.message.core;

import java.io.File;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import com.sduept.bigdata.fault.system.properties.FaultSystemProperties;
import com.sduept.cim.model.extend.ProtectWaveFile;
import com.sduept.cim.model.extend.WaveFile;
import com.sduept.cim.model.protection.ProtectionEquipment;
import com.sduept.cim.model.protection.WaveEquipment;
import com.sduept.cim.wave.service.CfgConfigureManager;
import com.sduept.cim.wave.service.DmfCreateManager;
import com.sduept.cim.wave.service.ProtectCfgConfigureManager;
import com.sduept.common.tool.FileOpt;
import com.sduept.core.dao.SuperDAO;
import com.sduept.generator.ObjectId;
import com.sduept.stardfile.parser.model.cfg.CfgFile;
import com.sduept.stardfile.parser.model.cfg.CfgTimeInfo;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

@Slf4j
@Service
public class WaveFileService {
	@Autowired
	private SuperDAO dao;
	@Autowired
	private CfgConfigureManager cm;
	@Autowired
	private ProtectCfgConfigureManager pm;
	@Autowired
	private DmfCreateManager dmfManager;
	@Autowired
	private FaultSystemProperties pros;
	private ExecutorService executor = Executors.newFixedThreadPool(10);

	private Object lbwjLock = new Object();
	private Object bhwjLock = new Object();

	@KafkaListener(topics = { "sf_bx_wave_download" })
	public void onMessage(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			log.info("接收到消息：sf_bx_wave_download");
			executor.submit(new Runnable() {
				@Override
				public void run() {
					JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
					String msgContent = jo.getString("content");
					persisitCentralizedWaveFiles(msgContent);
				}
			});
		}
	}

	@KafkaListener(topics = { "sf_bx_wave_download_bh" })
	public void onMessage1(ConsumerRecord<?, ?> record) {
		log.info("接收到消息：sf_bx_wave_download_bh");
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			executor.submit(new Runnable() {
				@Override
				public void run() {
					JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
					String msgContent = jo.getString("content");
					persisitProtectWaveFiles(msgContent);
				}
			});
		}

	}

	/**
	 * 检测集中录波cfg是否配置，如果没有配置，生成cfg
	 * 
	 * @param we
	 */
	private void checkFaultEquipmentCfgConfig(WaveEquipment we, WaveFile waveFile) {
		cm.checkChannels(we, waveFile);
	}

	/**
	 * 检测集中录波dmf是否配置，如果没有配置，生成dmf
	 * 
	 * @param we
	 */
	private void checkFaultEquipmentDmfConfig(WaveEquipment we) {
		if (!StringUtils.isEmpty(we.getCfg()) && StringUtils.isEmpty(we.getDmf())) {
			try {
				dmfManager.createDmf(we);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param filePath  cfg文件path
	 * @param stationId
	 * @param equId
	 * @param equName
	 */
	public void persisitProtectWaveFiles(String messageContent) {
		// 防止多线情况下的并发
		synchronized (bhwjLock) {
			log.info("接收到保护录波文件："+messageContent);
			List<ProtectWaveFile> list = new ArrayList<ProtectWaveFile>();
			JSONArray ja = JSONArray.fromObject(messageContent);
			Iterator<String> it = ja.iterator();
			while (it.hasNext()) {
				String str = it.next().toString();
				try {
					String[] paths = str.split(File.separator);
					String filePath = pros.getWaveFilePath() + str + ".cfg";
					String equId = null;
					String equName = null;
					String stationName = null;
					String equType = null;
					if (paths.length > 1) {
						equId = paths[1];
						ProtectionEquipment pe = dao.findById(equId, ProtectionEquipment.class);
						if (pe != null) {
							pm.checkProtectEquipmentCfgConfig(pe);
							pm.checkProtectEquipmentDmfConfig(pe);
							equName = pe.getName();
							stationName = pe.getSubName();
							equType = pe.getType();
							if (StringUtils.isEmpty(equType)) {
								String protectedType = pe.getProtectedType();
								if ("断路器".equals(protectedType))
									equType = "断路器保护";
							}
						}
					}
					File f = new File(filePath);
					if (!f.exists()) {
						f = new File(pros.getWaveFilePath() + str + ".CFG");
						if (!f.exists()) {
							continue;
						}
					}
					if (f.length() > 0) {
						List<ProtectWaveFile> tlist = dao.findByCondition(ProtectWaveFile.class, "path=?", str);
						if (tlist.size() == 0) {
							ProtectWaveFile wf = new ProtectWaveFile();
							wf.setId(ObjectId.get().toString());
							wf.setStationName(stationName);
							wf.setEquipment(equName);
							wf.setEquipId(equId);
							wf.setEquipType(equType);
							wf.setPath(str);
							wf.setAnalysisStatus(0);
							String[] dateInfos = getWaveFileDateInfo(filePath);
							wf.setStartSecond(dateInfos[0]);
							wf.setStartMicro(dateInfos[1]);
							wf.setTriggerSecond(dateInfos[2]);
							wf.setTriggerMicro(dateInfos[3]);
							list.add(wf);
						}
					}
				} catch (Exception e) {
					log.error("文件解析出错，地址为：" + str);
					e.printStackTrace();
				}
			}
			try {
				dao.persistentBatch(ProtectWaveFile.class, list);
				log.info("更新录播文件存档："+list.size());
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param filePath  cfg文件path
	 * @param stationId
	 * @param equId
	 * @param equName
	 */
	private void persisitCentralizedWaveFiles(String messageContent) {
		// 防止多线情况下的并发
		synchronized (lbwjLock) {
			log.info("接收到录波文件："+messageContent);
			List<WaveFile> list = new ArrayList<WaveFile>();
			JSONArray ja = JSONArray.fromObject(messageContent);
			Iterator<String> it = ja.iterator();
			while (it.hasNext()) {
				String str = it.next().toString();
				String[] paths = str.split(Matcher.quoteReplacement(File.separator));
				String filePath = pros.getWaveFilePath() + str + ".cfg";
				String equId = null;
				String equName = null;
				String stationName = null;
				WaveEquipment we = null;
				if (paths.length > 1) {
					equId = paths[1];
					we = dao.findById(equId, WaveEquipment.class);
					if (we != null) {
						equName = we.getName();
						stationName = we.getSubName();
					}
				}
				File f = new File(filePath);
				if (!f.exists()) {
					f = new File(pros.getWaveFilePath() + str + ".CFG");
					if (!f.exists()) {
						continue;
					}
				}
				if (f.length() > 0) {
					List<WaveFile> tlist = dao.findByCondition(WaveFile.class, "path=?", str);
					if (tlist.size() == 0) {
						WaveFile wf = new WaveFile();
						wf.setId(ObjectId.get().toString());
						wf.setStationName(stationName);
						wf.setEquipment(equName);
						wf.setEquipId(equId);
						wf.setEquipType("录波器");
						wf.setPath(str);
						wf.setAnalysisStatus(0);
						String[] dateInfos = getWaveFileDateInfo(filePath);
						wf.setStartSecond(dateInfos[0]);
						wf.setStartMicro(dateInfos[1]);
						wf.setTriggerSecond(dateInfos[2]);
						wf.setTriggerMicro(dateInfos[3]);
						list.add(wf);
						if (null != we) {
							try {
								checkFaultEquipmentCfgConfig(we, wf);
								checkFaultEquipmentDmfConfig(we);
							} catch (Exception e) {
								log.error("文件比较出错，地址为：" + str);
							}
						}
					}
				}
			}
			try {
				dao.persistentBatch(WaveFile.class, list);
				log.info("更新录波文件列表："+list.size());
			} catch (SQLException e) {
				e.printStackTrace();
			}

		}
	}

	/**
	 * @param path
	 * @return
	 */
	private String[] getWaveFileDateInfo(String path) {
		String[] result = new String[4];
		try {
			String cfgFileContent = FileOpt.readToString(path, "gbk");
			CfgFile cfgFile = new CfgFile();
			cfgFile.ReadCfgFile(cfgFileContent);
			CfgTimeInfo[] cti = cfgFile.timeInfos;
			if (cti.length == 2) {
				String startSecond = cti[0].yyyy + "-" + (cti[0].MM < 10 ? "0" + cti[0].MM : cti[0].MM) + "-"
						+ (cti[0].dd < 10 ? "0" + cti[0].dd : cti[0].dd) + " "
						+ (cti[0].hh < 10 ? "0" + cti[0].hh : cti[0].hh) + ":"
						+ (cti[0].mm < 10 ? "0" + cti[0].mm : cti[0].mm) + ":"
						+ (cti[0].ss < 10 ? "0" + cti[0].ss : cti[0].ss);
				String startMicro = cti[0].ssssss + "";
				String triggerSecond = cti[1].yyyy + "-" + (cti[1].MM < 10 ? "0" + cti[1].MM : cti[1].MM) + "-"
						+ (cti[1].dd < 10 ? "0" + cti[1].dd : cti[1].dd) + " "
						+ (cti[1].hh < 10 ? "0" + cti[1].hh : cti[1].hh) + ":"
						+ (cti[1].mm < 10 ? "0" + cti[1].mm : cti[0].mm) + ":"
						+ (cti[1].ss < 10 ? "0" + cti[1].ss : cti[1].ss);
				String triggerMicro = cti[1].ssssss + "";
				result[0] = startSecond;
				result[1] = startMicro;
				result[2] = triggerSecond;
				result[3] = triggerMicro;
			}
		} catch (Exception e) {
			log.error("录波文件读取出错，文件地址为：" + path);
			e.printStackTrace();
		}
		return result;
	}
}
